package cloud.xiguapi.ubas.analysis.pv.model;

import cloud.xiguapi.ubas.model.PageViewCount;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 09:04
 * desc:
 */
public class FilePVCountProcess extends KeyedProcessFunction<Long, PageViewCount, PVResult> {

    /**
     * 定义状态, 保存当前的总count值
     */
    private transient ValueState<Long> totalCountState;

    @Override
    public void open(Configuration parameters) throws Exception {
        // this.totalCountState = getRuntimeContext().getState(new ValueStateDescriptor<>("total-count", Long.class));
        // 初始化状态
        this.totalCountState = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("total-count", Long.class, 0L));
    }

    @Override
    public void processElement(PageViewCount value, Context ctx, Collector<PVResult> out) throws Exception {
        // 累加
        totalCountState.update(totalCountState.value() + value.getCount());
        // 注册定时器
        ctx.timerService().registerEventTimeTimer(value.getWindowEnd() + 1);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<PVResult> out) throws Exception {
        // 定时器触发, 所有分组count值都到齐, 直接输出当前的总count数量
        Long totalCount = totalCountState.value();
        // out.collect(new PageViewCount("pv", ctx.getCurrentKey(), totalCount));

        out.collect(PVResult.builder()
                .windowEnd(new Timestamp(ctx.getCurrentKey()).toString())
                .count(totalCount)
                .build());

        // 清空状态
        totalCountState.clear();
    }
}

